package com.xiaochong.bitong.job.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.bitong.job.config.hbase.CurrencyInfoHbaseComponent;
import com.xiaochong.bitong.job.config.hbase.HbaseBaseQueryComponent;
import com.xiaochong.service.common.config.redis.RedisServer;
import com.xiaochong.tir.common.data.hbasepo.TradesToKLine;
import com.xiaochong.tir.common.data.mapper.ExchangeInfoMapper;
import com.xiaochong.tir.common.data.mapper.ExchangeTransactionpairMiddleMapper;
import com.xiaochong.tir.common.data.mapper.FundsFlowMapper;
import com.xiaochong.tir.common.data.mapper.KLineMapper;
import com.xiaochong.tir.common.data.po.KLine;
import com.xiaochong.tir.common.util.*;
import net.sf.json.util.JSONTokener;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.DayOfWeek;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Map;

/**
 * @Description: k线舒适化
 * @Author: org
 * @CreateDate: 2018/8/30 15:11
 */
@Service
public class KActionInitDataService {
    @Autowired
    private ExchangeInfoMapper exchangeInfoMapper;
    @Autowired
    private ExchangeTransactionpairMiddleMapper exchangeTransactionpairMiddleMapper;
    @Autowired
    private HbaseBaseQueryComponent hbaseBaseQueryComponent;
    @Autowired
    private CurrencyInfoHbaseComponent currencyInfoHbaseComponent;
    @Autowired
    private FundsFlowMapper fundsFlowMapper;
    @Autowired
    private RedisServer redisServer;
    @Autowired
    private KLineMapper kLineDao;

    public String KActionInitData(String kType) {
        LogTrace.beginTrace("初始化数据启动");
        if(StringUtils.isBlank(kType)){
            return  "fail:请传参数";
        }
        LocalDateTime nowDateTime = LocalDateTime.now();
        boolean flag = true;
        if("5M".equals(kType)){
            flag = MakeMKActionPart(5, HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_5M,nowDateTime,DateUtils.format_ymdhm,"5M");
        }else if("15M".equals(kType)){
            flag = MakeMKActionPart(15, HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_15M,nowDateTime,DateUtils.format_ymdhm,"15M");
        }else if("30M".equals(kType)){
            flag = MakeMKActionPart(30, HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_30M,nowDateTime,DateUtils.format_ymdhm,"30M");
        }else if("1H".equals(kType)){
            flag = MakeMKActionPart(60, HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1H,nowDateTime,DateUtils.format_ymdh,"1H");
        }else if("4H".equals(kType)){
            flag = MakeMKActionPart(240, HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_4H,nowDateTime,DateUtils.format_ymdh,"4H");
        }else{
            return  "fail:kType参数错误";
        }
        if(flag){
            return  "success:"+kType;
        }else{
            return  "fail:"+kType;
        }
    }
    public  boolean MakeMKActionPart(int m, String rowHead, LocalDateTime nowDateTime, DateTimeFormatter format,String type) {
        //循环交易所交易对
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
        if(StringUtils.isBlank(dataString)){
            return false;
        }
        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        LocalDateTime nowDate = LocalDateTime.now();
        if("1H".equals(type) ){//得到当前时间段
            int yu = nowDate.getMinute()% 60 ;
            nowDate = nowDate.minusMinutes(yu);
        }else if("4H".equals(type)){
            int yu = nowDate.getMinute()% 60 ;
            nowDate = nowDate.minusMinutes(yu);
            yu = nowDate.getHour()%4;
            nowDate = nowDate.minusHours(yu);
        }else{
            int yu = nowDate.getMinute()% m ;
            nowDate = nowDate.minusMinutes(yu);
        }
        for (String key:exchange.keySet()) {
          /*  if(Integer.valueOf(key).intValue() != 16){
                continue;
            }*/
            try {
                JSONObject jsonObject = exchange.getJSONObject(key);
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    //生成k线
                    MakeMKAction(m,rowHead,Integer.valueOf(key),symbol,nowDate,format);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return true;
    }
    public  boolean MakeMKActionOfExchange(String type,LocalDateTime nowDateTime) {
        //循环交易所交易对
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
        if(StringUtils.isBlank(dataString)){
            return false;
        }
        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        while (nowDateTime.getMinute()%60 != 0){
            nowDateTime = nowDateTime.minusMinutes(1);
        }
        for (String key:exchange.keySet()) {
            LogTrace.infoKvs("一分钟K线数据", "key", key);
            /*if(Integer.valueOf(key).intValue() != 16){
                continue;
            }*/
            try {
                JSONObject jsonObject = exchange.getJSONObject(key);
                String exchangeName = (String) jsonObject.get("name");
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    if("4H".equals(type)){
                        LocalDateTime dateDate = nowDateTime.minusMinutes(0);
                        while(dateDate.getHour()%4 !=0){
                            dateDate = dateDate.minusHours(1);
                        }
                        make4HKactionData(Integer.valueOf(key),symbol,dateDate);
                    }else if("1W".equals(type)){
                        LocalDateTime dateDate = nowDateTime.minusMinutes(0);
                        while (dateDate.getDayOfWeek() != DayOfWeek.SUNDAY){
                            dateDate = dateDate.minusDays(1);
                        }
                        for(int d=0;d<30;d++){
                            dateDate = dateDate.minusDays(7);
                            makeWeekMonthData(7 , key, symbol, dateDate,KLineTypeEnum.WEEK, exchangeName);
                        }
                    }else if("1MON".equals(type)){
                        LocalDateTime dateDate = nowDateTime.minusMinutes(0);
                        while (dateDate.getDayOfMonth() != 1){
                            dateDate = dateDate.minusDays(1);
                        }
                        for(int d=0;d<6;d++){
                            dateDate = dateDate.minusMonths(1);
                            int monthDay = DateUtils.getDayOfMonth(dateDate.getYear(),dateDate.getMonthValue());//获取本月天数
                            makeWeekMonthData(monthDay , key, symbol, dateDate,KLineTypeEnum.MONTH, exchangeName);
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return true;
    }

    /**
     * 生成周k和月k数据
     * @param key
     * @param symbol
     * @param dateDate
     * @param week
     */
    public void makeWeekMonthData(int day ,String key,String symbol,LocalDateTime dateDate,KLineTypeEnum week,String exchangeName){
        try {
            //分别查询day天前的数据，日k，然后判断最高最低，最后保存数据库中
            KLine kLine = new KLine();
            LocalDateTime oldLocal = dateDate.minusMinutes(0);
            //循环获取day天日k数据
            String openPriceStr;
            String highOpenStr;
            String lowCloseStr;
            String closePriceStr;
            String volumeStr;
            for(int z=0;z<day;z++){
                String datastr = DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd);
                KLine kLineQuery1 = new KLine();
                kLineQuery1.setKDate(datastr);
                kLineQuery1.setKType(KLineTypeEnum.DAY.getIndex().toString());
                kLineQuery1.setExchangeId(Integer.valueOf(key));
                kLineQuery1.setTransationPairName(symbol);
                kLineQuery1 = kLineDao.selectByEntity(kLineQuery1);
                if(kLineQuery1 != null){
                    if(StringUtils.isBlank(kLineQuery1.getOpenPrice())){
                        continue;
                    }
                    if(StringUtils.isBlank(kLineQuery1.getHighOpen())){
                        continue;
                    }
                    if(StringUtils.isBlank(kLineQuery1.getLowClose())){
                        continue;
                    }
                    if(StringUtils.isBlank(kLineQuery1.getClosePrice())){
                        continue;
                    }
                    openPriceStr = kLineQuery1.getOpenPrice();
                    highOpenStr = kLineQuery1.getHighOpen();
                    lowCloseStr = kLineQuery1.getLowClose();
                    closePriceStr = kLineQuery1.getClosePrice();
                    volumeStr = kLineQuery1.getVolume()!=null?kLineQuery1.getVolume():"0";
                    //交易量
                    BigDecimal volume = new BigDecimal(kLine.getVolume()!=null? kLine.getVolume():"0");
                    volume = volume.add(new BigDecimal(volumeStr));
                    kLine.setVolume(volume.toString());
                    //开盘
                    if (StringUtils.isBlank(kLine.getOpenPrice())) {
                        kLine.setOpenPrice(openPriceStr);
                    }
                    //收盘
                    kLine.setClosePrice(closePriceStr);
                    //最高价格
                    if (StringUtils.isBlank(kLine.getHighOpen())) {
                        kLine.setHighOpen(highOpenStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                        BigDecimal highOpen = new BigDecimal(kLine.getHighOpen());
                        if (bigDecimal.compareTo(highOpen) > 0) {
                            kLine.setHighOpen(highOpenStr);
                        }
                    }
                    //最低价格
                    if (StringUtils.isBlank(kLine.getLowClose())) {
                        kLine.setLowClose(lowCloseStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                        BigDecimal lowClose = new BigDecimal(kLine.getLowClose());
                        if (bigDecimal.compareTo(lowClose) < 0) {
                            kLine.setLowClose(lowCloseStr);
                        }
                    }
                }
                oldLocal = oldLocal.plusDays(1);
            }
            //插入或者更新
            if(kLine != null && StringUtils.isNotBlank(kLine.getOpenPrice()) && StringUtils.isNotBlank(kLine.getClosePrice())) {
                //oldLocal = dateDate.minusDays(day);
                KLine kLine1 = new KLine();
                kLine1.setKDate(DateUtils.localDateTimeFormatString(dateDate, DateUtils.format_ymd));
                kLine1.setKType(week.getIndex().toString());
                kLine1.setExchangeId(Integer.valueOf(key));
                kLine1.setTransationPairName(symbol);
                kLine1 = kLineDao.selectByEntity(kLine1);
                //如果存在更细，否则直接插入
                if (kLine1 == null) {
                    kLine.setKName(week.getName());
                    kLine.setExchangeName(exchangeName);
                    kLine.setKDate(DateUtils.localDateTimeFormatString(dateDate, DateUtils.format_ymd));
                    kLine.setKType(week.getIndex().toString());
                    kLine.setExchangeId(Integer.valueOf(key));
                    kLine.setTransationPairName(symbol);
                    kLine.setCreateTime(new Date());
                    kLineDao.insert(kLine);
                } else {
                    kLine1.setExchangeName(exchangeName);
                    kLine1.setOpenPrice(kLine.getOpenPrice());
                    kLine1.setHighOpen(kLine.getHighOpen());
                    kLine1.setLowClose(kLine.getLowClose());
                    kLine1.setClosePrice(kLine.getClosePrice());
                    kLine1.setUpdateTime(new Date());
                    kLine1.setKName(week.getName());
                    kLineDao.updateByPrimaryKey(kLine1);
                }
            }

        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("时时周k、月k-error",e);
        }
    }
    /**
     * 4H数据
     * @param exchangeId
     * @param transactionName
     * @param time
     */
    public void make4HKactionData(Integer exchangeId, String transactionName, LocalDateTime time){
        for(int i=0;i<500;i++){
            time = time.minusHours(4);
            calculateFourHMakeKAction(exchangeId, transactionName, time);
        }
    }
    /**
     * 4h的k线图
     * @param exchangeId
     * @param transactionName
     * @param time
     * @throws Exception
     */
    private void calculateFourHMakeKAction(Integer exchangeId, String transactionName, LocalDateTime time){
        try {
           // LocalDateTime oldStart = time.minusMinutes(0);
            TradesToKLine tradesToKLineNew = new TradesToKLine();
            tradesToKLineNew.setVolume("0");
            tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(time).getTime()));
            // oldStart = time.minusMinutes(60);
            for(int i=0;i<4;i++){
                String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time, DateUtils.format_ymdh);
                //LogTrace.infoKvs("4hK线数据", "rowKey", rowKey);
                String openPriceStr;
                String highOpenStr;
                String lowCloseStr;
                String closePriceStr;
                String volumeStr;
                TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKey);
                if(tradesToKLine != null){
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                    //交易量
                    BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                    volume = volume.add(new BigDecimal(volumeStr));
                    tradesToKLineNew.setVolume(volume.toString());
                    //开盘
                    if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                        tradesToKLineNew.setOpenPrice(openPriceStr);
                    }
                    //收盘
                    tradesToKLineNew.setClosePrice(closePriceStr);
                    //最高价格
                    if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                        BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                        if (bigDecimal.compareTo(highOpen) > 0) {
                            tradesToKLineNew.setHighestPrice(highOpenStr);
                        }
                    }
                    //最低价格
                    if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                        BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                        if (bigDecimal.compareTo(lowClose) < 0) {
                            tradesToKLineNew.setLowestPrice(lowCloseStr);
                        }
                    }
                }
                time = time.plusHours(1);
            }
            if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
                String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_4H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time, DateUtils.format_ymdh);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
                LogTrace.infoKvs("时时数据初始化","rowKey",rowKey,"data",JSON.toJSONString(tradesToKLineNew));
                //System.out.println("rowKey:"+rowKey+",data:"+JSON.toJSONString(tradesToKLineNew));
            }

        }catch (Exception e){
            e.printStackTrace();
        }
    }
    /**
     * 获取hbase数据（根据rowkey获取1hk线数据）
     * @param rowKey
     * @return
     * @throws Exception
     */
    public TradesToKLine getHbaseDataByRowKey(String rowKey) throws Exception{
        Map<String, Object> hbaseData = hbaseBaseQueryComponent.selectData(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME
                , rowKey);
        if(null != hbaseData && hbaseData.get("volume") != null && hbaseData.get("openPrice") != null && hbaseData.get("highestPrice") != null&& hbaseData.get("lowestPrice") != null&& hbaseData.get("closePrice") != null){
            TradesToKLine tradesToKLine = new TradesToKLine();
            tradesToKLine.setTimestamp(hbaseData.get("timestamp").toString());
            tradesToKLine.setVolume(hbaseData.get("volume").toString());
            tradesToKLine.setOpenPrice(hbaseData.get("openPrice").toString());
            tradesToKLine.setClosePrice(hbaseData.get("closePrice").toString());
            tradesToKLine.setHighestPrice(hbaseData.get("highestPrice").toString());
            tradesToKLine.setLowestPrice(hbaseData.get("lowestPrice").toString());
            return tradesToKLine;
        }
        return null;
    }
    public void MakeMKAction(int m,String rowHead,Integer exchangeId,String transactionName,LocalDateTime time,DateTimeFormatter format){
        try {
            //循环500次数据
            String openPriceStr;
            String highOpenStr;
            String lowCloseStr;
            String closePriceStr;
            String volumeStr;
            LocalDateTime oldStart;
            TradesToKLine tradesToKLineNew;
            String data;
            for(int z=0;z<350;z++){
                oldStart = time.minusMinutes(0);
                tradesToKLineNew = new TradesToKLine();
                tradesToKLineNew.setVolume("0");
                tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(time).getTime()));
                for(int i=0;i<m;i++){
                    String rowKey = exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdhm);
                    tradesToKLineNew = compareKActionMinus(rowKey,tradesToKLineNew);
                    oldStart =oldStart.plusMinutes(1);
                }
                if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
                    String rowKey =rowHead+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time, format);
                    currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
                    LogTrace.infoKvs("初始化数据K线5M/15M/30M数据打印","rowkey",rowKey,"data",JSON.toJSONString(tradesToKLineNew));
                    //System.out.println("rowkey:"+rowKey+",data:"+JSON.toJSONString(tradesToKLineNew));
                }
                time = time.minusMinutes(m);
            }
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("初始化数据K线5M/15M/30M/1H/4H-Error",e);
        }
    }

    public Object kActionLongTimeNew(String kType, String beginTime, String endTime) {
        LogTrace.beginTrace("初始化数据启动");
        if(StringUtils.isBlank(kType) || StringUtils.isBlank(beginTime) || StringUtils.isBlank(endTime)){
            return  "fail:请传参数";
        }
        LocalDateTime nowDateTime = LocalDateTime.now();
        boolean flag = true;
        LogTrace.infoKvs("初始化数据K线日，周，月数据打印-kType","kType",kType);
        if("1D".equals(kType)){
            flag = MakeMKActionPartDayWeek("1D",HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1D,DateUtils.format_ymds,beginTime,endTime);
        }else if("1W".equals(kType)){
            flag = MakeMKActionPartDayWeek("1W",HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1W,DateUtils.format_ymds,beginTime,endTime);
        }else if("1MON".equals(kType)){
            flag = MakeMKActionPartDayWeek("1MON",HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1Mo,DateUtils.format_ymds,beginTime,endTime);
        }else{
            return  "fail:kType参数错误";
        }
        if(flag){
            return  "success:"+kType;
        }else{
            return  "fail:"+kType;
        }
    }

    /**
     *
     * @param type   1D,1W,1MON分别代表1天，一周，一月
     * @param head  保存hbase的rowkey头部标记
     * @param format  保存hbase的rowkey的日期，年月日、年月日时，年月日时分不同的k线使用不同的格式
     * @param beginTime  生产k线起始日期
     * @param endTime 生产k线结束日期
     * @param exchangeId 交易所id
     * @param transactionName 交易对名称
     * @param m 循环之后需要直接累加天数，日k累加一天，周k累加一周7天，月k累加一月30天
     * @return
     */
    private boolean MakeMKActionOfDayWeekMon(String type, String head, DateTimeFormatter format, String beginTime, String endTime,
                                             Integer exchangeId,String transactionName,int m) {
        LocalDateTime beginLocal = DateUtils.getDateTimeOfTimestamp( DateUtils.getLongTimeByFormat(beginTime+"0000",DateUtils.format_ymdhm));
        LocalDateTime endLocal = DateUtils.getDateTimeOfTimestamp( DateUtils.getLongTimeByFormat(endTime+"0000",DateUtils.format_ymdhm));
        if("1W".equals(type)) {
            while (beginLocal.getDayOfWeek() != DayOfWeek.SUNDAY){
                beginLocal = beginLocal.minusDays(1);
            }
        }else if("1MON".equals(type)){
            int yu = beginLocal.getDayOfMonth() - 1;
            beginLocal = beginLocal.minusDays(yu);
        }
        LogTrace.infoKvs("初始化数据K线日，周，月数据打印","type",type);
        while (beginLocal.isBefore(endLocal)){
            LocalDateTime tmp = beginLocal;
            TradesToKLine tradesToKLineNew = new TradesToKLine();
            tradesToKLineNew.setVolume("0");
            tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(beginLocal).getTime()));
            String rowKeyOld = head+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(beginLocal, format);
            if("1D".equals(type)){
                for(int i=0;i<1440;i++){
                    //获取当前分钟k线，比较最大值
                    String rowKey = exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(tmp, DateUtils.format_ymdhm);
                    tradesToKLineNew = compareKActionMinus(rowKey,tradesToKLineNew);
                    tmp = tmp.plusMinutes(1);
                }
            }else if("1W".equals(type)){
                for(int i=0;i<7;i++){
                    //获取当前分钟k线，比较最大值
                    String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1D+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(tmp, DateUtils.format_ymds);
                    tradesToKLineNew =  getHbaseDataByRowKey(rowKey,tradesToKLineNew);
                    tmp = tmp.plusDays(1);
                }
            }else if("1MON".equals(type)){
                //计算当前月份的天数
                m = DateUtils.getDayOfMonth(beginLocal.getYear(),beginLocal.getMonthValue());
                for(int i=0;i<m;i++){
                    //获取当前分钟k线，比较最大值
                    String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1D+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(tmp, DateUtils.format_ymds);
                    tradesToKLineNew =  getHbaseDataByRowKey(rowKey,tradesToKLineNew);
                    tmp = tmp.plusDays(1);
                }
            }
            //判断日k是否存在，存在保存
            if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKeyOld, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
                //System.out.println("rowkey:"+rowKeyOld+",data:"+JSON.toJSONString(tradesToKLineNew));
                LogTrace.infoKvs("初始化数据K线日，周，月数据打印","rowkey",rowKeyOld,"data",JSON.toJSONString(tradesToKLineNew));
            }
            //月k加一个月，其他加一天
            beginLocal = beginLocal.plusDays(m);
        }

        return  true;
    }
    public  boolean MakeMKActionPartDayWeek(String type, String rowHead,  DateTimeFormatter format,String beginTime, String endTime) {
        //循环交易所交易对
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
        if(StringUtils.isBlank(dataString)){
            return false;
        }
        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        for (String key:exchange.keySet()) {
            /*if(Integer.valueOf(key).intValue() != 16){
                continue;
            }*/
            try {
                int m =1;
                if("1D".equals(type)){
                    m = 1;
                }else if("1W".equals(type)){
                    m = 7;
                }else if("1MON".equals(type)){
                    m =30;
                }else{
                    return false;
                }
                JSONObject jsonObject = exchange.getJSONObject(key);
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    MakeMKActionOfDayWeekMon(type, rowHead,  format,  beginTime,  endTime,
                             Integer.valueOf(key), symbol,m);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return true;
    }
    public TradesToKLine compareKActionMinus(String rowKey,TradesToKLine tradesToKLineNew){
        try {
            String openPriceStr;
            String highOpenStr;
            String lowCloseStr;
            String closePriceStr;
            String volumeStr;
            LocalDateTime oldStart;
            String data = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                    , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                    , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                    , rowKey);
            if(StringUtils.isNotBlank(data)) {
                //下标： 0 时间戳 1 开盘 2最高价 3最低  4收盘 5交易量 6涨幅
                //  LogTrace.infoKvs("一分钟K线数据", "rowKey", rowKey, "data", data);
                Object listArray = new JSONTokener(data).nextValue();
                if (listArray instanceof net.sf.json.JSONArray) {
                    JSONArray jsonArray = JSON.parseArray(data);
                    openPriceStr = jsonArray.getString(1);
                    highOpenStr = jsonArray.getString(2);
                    lowCloseStr = jsonArray.getString(3);
                    closePriceStr = jsonArray.getString(4);
                    volumeStr = jsonArray.getString(5);

                } else {
                    JSONObject jsonObject3 = JSON.parseObject(data);
                    TradesToKLine tradesToKLine = jsonObject3.toJavaObject(TradesToKLine.class);
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                }

                //交易量
                BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                volume = volume.add(new BigDecimal(volumeStr));
                tradesToKLineNew.setVolume(volume.toString());

                //开盘
                if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                    tradesToKLineNew.setOpenPrice(openPriceStr);
                }
                //收盘
                tradesToKLineNew.setClosePrice(closePriceStr);
                //最高价格
                if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                    tradesToKLineNew.setHighestPrice(highOpenStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                    BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                    if (bigDecimal.compareTo(highOpen) > 0) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    }
                }
                //最低价格
                if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                    tradesToKLineNew.setLowestPrice(lowCloseStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                    BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                    if (bigDecimal.compareTo(lowClose) < 0) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    }
                }
            }

        }catch (Exception e){
            e.printStackTrace();
        }
        return tradesToKLineNew;
    }

    /**
     * 查询当前k线并且比较当前与上一次的最大值
     * @param rowKey  当前rowkey
     * @param kLine  上一次数据
     * @return
     * @throws Exception
     */
    public TradesToKLine getHbaseDataByRowKey(String rowKey,TradesToKLine kLine){
        try {
            Map<String, Object> hbaseData = hbaseBaseQueryComponent.selectData(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME
                    , rowKey);
            if(null != hbaseData && hbaseData.get("volume") != null && hbaseData.get("openPrice") != null && hbaseData.get("highestPrice") != null&& hbaseData.get("lowestPrice") != null&& hbaseData.get("closePrice") != null){
                TradesToKLine tradesToKLine = new TradesToKLine();
                tradesToKLine.setTimestamp(hbaseData.get("timestamp").toString());
                tradesToKLine.setVolume(hbaseData.get("volume").toString());
                tradesToKLine.setOpenPrice(hbaseData.get("openPrice").toString());
                tradesToKLine.setClosePrice(hbaseData.get("closePrice").toString());
                tradesToKLine.setHighestPrice(hbaseData.get("highestPrice").toString());
                tradesToKLine.setLowestPrice(hbaseData.get("lowestPrice").toString());
                if(tradesToKLine != null){
                    String openPriceStr = tradesToKLine.getOpenPrice();
                    String highOpenStr = tradesToKLine.getHighestPrice();
                    String lowCloseStr = tradesToKLine.getLowestPrice();
                    String closePriceStr = tradesToKLine.getClosePrice();
                    String volumeStr = tradesToKLine.getVolume();
                    //交易量
                    BigDecimal volume = new BigDecimal(kLine.getVolume());
                    volume = volume.add(new BigDecimal(volumeStr));
                    kLine.setVolume(volume.toString());
                    //开盘
                    if (StringUtils.isBlank(kLine.getOpenPrice())) {
                        kLine.setOpenPrice(openPriceStr);
                    }
                    //收盘
                    kLine.setClosePrice(closePriceStr);
                    //最高价格
                    if (StringUtils.isBlank(kLine.getHighestPrice())) {
                        kLine.setHighestPrice(highOpenStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                        BigDecimal highOpen = new BigDecimal(kLine.getHighestPrice());
                        if (bigDecimal.compareTo(highOpen) > 0) {
                            kLine.setHighestPrice(highOpenStr);
                        }
                    }
                    //最低价格
                    if (StringUtils.isBlank(kLine.getLowestPrice())) {
                        kLine.setLowestPrice(lowCloseStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                        BigDecimal lowClose = new BigDecimal(kLine.getLowestPrice());
                        if (bigDecimal.compareTo(lowClose) < 0) {
                            kLine.setLowestPrice(lowCloseStr);
                        }
                    }
                }
                //return kLine;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return kLine;
    }
}
